package com.duoec.dw.lineage.service.impl;

import com.duoec.dw.lineage.dto.*;
import com.duoec.dw.lineage.enums.DatabaseTypeEnum;
import com.duoec.dw.lineage.exceptions.LineageBadSqlException;
import com.duoec.dw.lineage.exceptions.LineageTableNotFoundException;
import com.duoec.dw.lineage.service.LineageTableService;
import com.duoec.dw.lineage.service.impl.desc.*;
import com.duoec.dw.lineage.service.impl.desc.field.TableDescFieldProcessor;
import com.duoec.dw.lineage.utils.DateTool;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.BadSqlGrammarException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author xuwenzhen
 */
@Service
public class LineageTableServiceDbImpl implements LineageTableService {
    private static final Logger logger = LoggerFactory.getLogger(LineageTableServiceDbImpl.class);
    private static final String STR_COL_NAME = "col_name";
    private static final String STR_SHARP = "#";
    private static final String STR_DESC = "desc formatted ";
    public static final String STR_SHARP_COL_NAME = "# col_name";
    public static final String STR_SHARP_PARTITION_INFORMATION = "# Partition Information";
    public static final String STR_SHARP_DETAILED_TABLE_INFORMATION = "# Detailed Table Information";
    public static final String STR_SHARP_STORAGE_INFORMATION = "# Storage Information";
    public static final String STR_TABLE_NOT_FOUND = "Table not found";

    private static final String TABLE_TYPE = "Table Type:";
    private static final String LAST_MODIFIED_TIME = "param.last_modified_time";
    private static final String LAST_MODIFIED_BY = "param.last_modified_by";
    private static final String LOCATION = "Location:";
    private static final String OWNER = "Owner:";
    private static final String CREATE_TIME = "CreateTime:";
    private static final String TOTAL_SIZE = "param.totalSize";
    private static final String TABLE_FILE_TYPE = "InputFormat:";


    @Autowired
    @Qualifier("hiveJdbcTemplate")
    private JdbcTemplate hiveJdbcTemplate;

    @Autowired
    private List<TableDescTypeProcessor> processorList;

    private Map<String, TableDescTypeProcessor> processorMap;

    @Autowired
    private List<TableDescFieldProcessor> tableDescFieldProcessorList;

    private Map<String, Map<String, TableDescFieldProcessor>> tableDescFieldProcessorMap = Maps.newHashMap();

    @PostConstruct
    public void init() {
        processorMap = processorList.stream().collect(Collectors.toMap(TableDescTypeProcessor::getType, item -> item));

        tableDescFieldProcessorList.forEach(process -> {
            Map<String, TableDescFieldProcessor> processMap = tableDescFieldProcessorMap.computeIfAbsent(process.getTypeName(), key -> Maps.newHashMap());
            processMap.put(process.getFieldName(), process);
        });
    }

    /**
     * 获取表字段信息
     *
     * @param tableCoordinate 数据表
     * @return 字段信息列表
     */
    @Override
    public TableInfoDto getTableDto(TableCoordinateDto tableCoordinate) {
        if (tableCoordinate.getDatabaseName() == null && tableCoordinate.getTableName() == null) {
            return null;
        }

        TableInfoDto tableInfo = LineageTableService.getTable(tableCoordinate.toString());
        if (tableInfo != null) {
            return tableInfo;
        }

        tableInfo = new TableInfoDto();
        tableInfo.setDatabaseType(DatabaseTypeEnum.HIVE.getValue());
        tableInfo.setDatabaseName(tableCoordinate.getDatabaseName());
        tableInfo.setTableName(tableCoordinate.getTableName());
        TableDescReaderState state = new TableDescReaderState();
        List<TableDescItem> fields;
        try {
            fields = getTableDescItems(tableCoordinate, state);
        } catch (BadSqlGrammarException e) {
            if (e.getMessage() != null && e.getMessage().contains(STR_TABLE_NOT_FOUND)) {
                throw new LineageTableNotFoundException("找不到表：" + tableCoordinate.toString(), e);
            }
            throw new LineageBadSqlException(e.getMessage(), e);
        }

        Iterables.removeIf(fields, Objects::isNull);
        setFields(tableInfo, fields);
        setDetailedTableInfo(tableInfo, fields);
        setStorageTableInfo(tableInfo, fields);
        return tableInfo;
    }

    private List<TableDescItem> getTableDescItems(TableCoordinateDto tableCoordinate, TableDescReaderState state) throws BadSqlGrammarException {
        return hiveJdbcTemplate.query(STR_DESC + tableCoordinate.toString(), (rs, rowNum) -> {
            String type = state.getType();
            if (type == null) {
                type = FieldTableDescTypeProcessor.TYPE;
                state.setType(type);
            }
            String colName = rs.getString(STR_COL_NAME);
            if (colName.startsWith(STR_SHARP)) {
                setTableDescType(state, colName);
                return null;
            }

            TableDescTypeProcessor processor = processorMap.get(type);
            if (processor == null) {
                logger.error("未实现{}的解析器！", colName);
                return null;
            }
            return processor.process(state, rs);
        });
    }

    private void setDetailedTableInfo(TableInfoDto dto, List<TableDescItem> fields) {
        Map<String, TableDescItem> tableFieldMap = fields.stream()
                .filter(field -> DetailTableDescTypeProcessor.TYPE.equals(field.getType()))
                .collect(Collectors.toMap(TableDescItem::getColumnName, field -> field));

        TableDescItem tableType = tableFieldMap.get(TABLE_TYPE);
        if (tableType != null) {
            dto.setTableType(tableType.getDataType());
        }
        TableDescItem lastModifiedTime = tableFieldMap.get(LAST_MODIFIED_TIME);
        if (lastModifiedTime != null && StringUtils.isNumeric(lastModifiedTime.getDataType())) {
            dto.setLastModifiedTime(DateTool.getDateTimeOfSec(Long.valueOf(lastModifiedTime.getDataType())));
        }
        TableDescItem lastModifiedBy = tableFieldMap.get(LAST_MODIFIED_BY);
        if (lastModifiedBy != null) {
            dto.setLastModifiedBy(lastModifiedBy.getDataType());
        }
        TableDescItem location = tableFieldMap.get(LOCATION);
        if (location != null) {
            dto.setLocation(location.getDataType());
        }
        TableDescItem owner = tableFieldMap.get(OWNER);
        if (owner != null) {
            dto.setOwner(owner.getDataType());
        }
        TableDescItem createTime = tableFieldMap.get(CREATE_TIME);
        if (createTime != null) {
            DateTimeFormatter df = DateTimeFormatter.ofPattern("EEE MMM dd HH:mm:ss z yyyy", Locale.US);
            dto.setTableCreateTime(LocalDateTime.parse(createTime.getDataType(), df));
        }
        // 元数据取出来的是Bytes级别的数据,这里直接保存，前端展示的时候进行数据的单位转换
        TableDescItem totalSize = tableFieldMap.get(TOTAL_SIZE);
        if (totalSize != null && StringUtils.isNumeric(totalSize.getDataType())) {
            dto.setTableStorageCapacity(Long.valueOf(totalSize.getDataType()));
        }
    }

    private void setStorageTableInfo(TableInfoDto dto, List<TableDescItem> fields) {
        Map<String, TableDescItem> tableFieldMap = fields.stream()
                .filter(field -> StorageTableDescTypeProcessor.TYPE.equals(field.getType()))
                .collect(Collectors.toMap(TableDescItem::getColumnName, field -> field));
        TableDescItem tableFileType = tableFieldMap.get(TABLE_FILE_TYPE);
        if (tableFileType != null) {
            dto.setTableFileType(tableFileType.getDataType());
        }
    }

    private void setFields(TableInfoDto dto, List<TableDescItem> fields) {
        LinkedHashSet<TableFieldDto> tableFields = fields.stream()
                .filter(field -> FieldTableDescTypeProcessor.TYPE.equals(field.getType()))
                .map(this::toTableFieldDto)
                .collect(Collectors.toCollection(LinkedHashSet::new));
        int index = 0;
        for (TableFieldDto field : tableFields) {
            field.setIndex(index);
            ++index;
        }
        dto.setFields(tableFields);
    }

    private TableFieldDto toTableFieldDto(TableDescItem item) {
        TableFieldDto field = new TableFieldDto();
        field.setColName(item.getColumnName());
        field.setDataType(item.getDataType());
        field.setComment(item.getComment());
        if (item.getPartition() != null && item.getPartition()) {
            //分区
            field.setFieldType(2);
        } else {
            field.setFieldType(1);
        }
        return field;
    }

    private void setTableDescType(TableDescReaderState state, String colName) {
        if (colName.startsWith(STR_SHARP_COL_NAME)) {
            //什么都不做
        } else if (colName.startsWith(STR_SHARP_PARTITION_INFORMATION)) {
            state.resetType(PartitionTableDescTypeProcessor.TYPE);
            state.setSubType(null);
        } else if (colName.startsWith(STR_SHARP_DETAILED_TABLE_INFORMATION)) {
            state.resetType(DetailTableDescTypeProcessor.TYPE);
            state.setSubType(null);
        } else if (colName.startsWith(STR_SHARP_STORAGE_INFORMATION)) {
            state.resetType(StorageTableDescTypeProcessor.TYPE);
            state.setSubType(null);
        } else {
            throw new RuntimeException("未处理字段：" + colName);
        }
    }
}
